#
# This file is a part of the open-eBackup project.
# This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
# If a copy of the MPL was not distributed with this file, You can obtain one at
# http://mozilla.org/MPL/2.0/.
#
# Copyright (c) [2024] Huawei Technologies Co.,Ltd.
#
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
#

import hashlib
import json
import locale
import os
import pathlib
import platform
import random
import re
import shutil
import signal
import stat
import time

import pexpect
from bs4 import BeautifulSoup
from pexpect import popen_spawn

from common.cleaner import clear
from common.common import check_command_injection, read_result_file, execute_cmd, get_host_sn
from common.common import output_execution_result, retry_when_exception, execute_cmd_list, write_content_to_file, \
    get_local_ips, is_windows
from common.const import RepositoryDataTypeEnum, CMDResult, ParamConstant, BackupTypeEnum
from common.err_code import CommErrCode
from common.exception.common_exception import ErrCodeException
from common.parse_parafile import get_env_variable
from common.security.anonym_utils.anonymity import Anonymity
from common.util.exec_utils import ExecFuncParam, su_exec_cmd_list
from oracle import logger
from oracle.common.constants import PexpectResult, ArchiveLogMode, InstanceStatus, PluginPathConstant, \
    exit_code_error_code_map, ErrorCode, DELETING_PATH_WHITE_LIST, DataBaseRole, ScriptExitCode, \
    OracleFilePathSql, Platform, TIME_FORMAT, CacheRepoFilename, TaskType, OracleResourceKeyName, ERROR_SQL_EXPECT
from oracle.common.user_env_common import get_asm_user_env_by_linux, get_oracle_user_env_by_linux, \
    build_user_env, get_user_shell_type
from oracle.common.windows_common import execute_mount_bind_windows, execute_windows_cmd

if platform.system().lower() != "windows":
    import pwd
    import grp
    from common.util.check_user_utils import check_os_user


def send_params_to_pexpect(params, child):
    """
    为shell脚本发送参数
    :param params: 参数字典
    :param child: pexpect.spawn(...)返回的操作程序操作句柄
    :return:
    """
    child.sendline(json.dumps(params))


def create_params_content_for_shell(params):
    params_content = ""
    for key, value in params.items():
        params_content = f"{params_content}{key}={value} "
    return params_content.strip()


def parse_backup_path(repositories: list):
    """
    解析数据仓信息
    :param repositories:仓库列表
    :return: 仓库地址
    """
    backup_dirs = dict()
    for repository in repositories:
        repository_type = repository.get("repositoryType")
        if repository_type == RepositoryDataTypeEnum.DATA_REPOSITORY:
            backup_dirs["data_repository"] = repository.get("path", [""])
        elif repository_type == RepositoryDataTypeEnum.CACHE_REPOSITORY:
            backup_dirs["cache_repository"] = repository.get("path", [""])
        elif repository_type == RepositoryDataTypeEnum.LOG_REPOSITORY:
            backup_dirs["log_repository"] = repository.get("path", [""])
        elif repository_type == RepositoryDataTypeEnum.META_REPOSITORY:
            backup_dirs["meta_repository"] = repository.get("path", [""])
    return backup_dirs


@retry_when_exception(retry_times=3, delay=3)
def write_tmp_json_file(file_path, job_id, context):
    """
    写临时json文件
    :param file_path: 文件路径
    :param job_id: 任务id
    :param context: 文件内容
    """
    if os.path.exists(file_path) and os.path.islink(file_path):
        logger.error(f"The path[{file_path}] is invalid, job id: {job_id}.")
        return
    if os.path.exists(file_path):
        if platform.system().lower() == "windows":
            ret, realpath = True, file_path
        else:
            ret, realpath = check_path_in_white_list(file_path)
            logger.info(f"Check path in white list result is: {ret}, file path is: {realpath}.")
        if ret:
            os.remove(realpath)
            logger.info(f"Remove file: {realpath}, job id: {job_id}.")
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    modes = stat.S_IWUSR | stat.S_IRUSR | stat.S_IXUSR
    with os.fdopen(os.open(file_path, flags, modes), 'w+') as out_file:
        out_file.write(json.dumps(context))


def check_path_in_white_list(path):
    """
    检查文件路径是否在删除白名单里
    :param path: 文件路径
    :return: 文件是否是删除白名单路径; 文件路径
    """
    try:
        real_path = os.path.realpath(path)
    except Exception:
        logger.exception(f"The path is invalid!")
        return False, path
    if check_command_injection(real_path):
        return False, path

    for path in DELETING_PATH_WHITE_LIST:
        if real_path.find(path) == 0:
            return True, real_path

    if f"{real_path}/" in DELETING_PATH_WHITE_LIST:
        return True, real_path

    return False, path


def add_file_content(extend_content, file_name, split_code):
    """
    组装脚本备份结果文件内容
    :param extend_content: 扩展字段
    :param file_name: 文件名
    :param split_code: 分隔符
    """
    if platform.system().lower() == "windows":
        encoding = locale.getdefaultlocale()[1]
        file_content = read_result_file(file_name, encoding=encoding)
    else:
        file_content = read_result_file(file_name)
    os.remove(file_name)
    logger.info(f"Delete tmp result file after read file_name={file_name}.")
    if len(file_content) > 0:
        for content in file_content.splitlines():
            # 当分隔符是等号的时候，取第一个等号进行分隔
            row = content.split(split_code, 1)
            extend_content[row[0]] = row[1]


def clear_repository_dir(file_path, log_id):
    """
    清空仓库文件
    :param file_path: 文件路径
    :param log_id: 日志id
    """
    clear_repository_dir_exclude_prefix(file_path, log_id, '')


def clear_repository_dir_exclude_prefix(file_path, log_id, exclude_file_prefix):
    """
    清空仓库文件
    :param file_path: 文件路径
    :param log_id: 日志id
    :param exclude_file_prefix: 排除删除的文件前缀
    """
    if platform.system().lower() == "windows":
        ret, realpath = True, file_path
    else:
        ret, realpath = check_path_in_white_list(file_path)
    if not ret:
        logger.info(f"File path is not in white list, reject to delete tmp dir: {file_path}, "
                    f"realpath: {realpath}, log_id: {log_id}.")
        return
    for path in os.listdir(realpath):
        new_path = os.path.join(realpath, path)
        if '.snapshot' in new_path:
            continue
        # 排除的文件不删除
        if exclude_file_prefix and CacheRepoFilename.RESTORE_POST_PROGRESS in new_path:
            continue
        if os.path.isfile(new_path):
            try:
                os.remove(new_path)
            except FileNotFoundError as exception:
                logger.warning(f"File is not exist, path is: {new_path}.")
        elif os.path.isdir(new_path):
            shutil.rmtree(new_path, ignore_errors=True)
    logger.info(f"The directory is deleted successfully, dir path is: {file_path}.")


def close_child(child):
    if child is None:
        return
    try:
        if is_windows():
            child.kill(signal.SIGTERM)
        else:
            child.close()
    except Exception as err:
        logger.warn(f"close child error,err:{err}")


def parse_html_result(exec_result):
    """
    解析Oracle结果产生的HTML结果
    :param exec_result: 执行结果
    :return: 返回Map的查询结果
    """
    soup = BeautifulSoup(exec_result, "html.parser")
    trs = soup.find_all(name="tr")
    params = []
    soup = BeautifulSoup(str(trs[0]), "html.parser")
    ths = soup.find_all(name='th', attrs={"scope": "col"})

    headers = [header.get_text().strip() for header in ths]
    for i in range(1, len(trs)):
        _soup = BeautifulSoup(str(trs[i]), "html.parser")
        tds = _soup.find_all(name='td')
        param = [td.get_text().strip() for td in tds]
        params.append(dict(zip(headers, param)))

    logger.info(f"Get parse html result: {params}.")
    return params


def parse_sql_result(cmd_result, key):
    """
    :param cmd_result: 字典结果
    :param key: 查询的数据库字段
    :return: 查询的字段结果
    """
    results = parse_html_result(cmd_result)
    if not results or not results[0].get(key):
        return ""
    return results[0].get(key)


def enter_password(child, db_user, db_password):
    """
    交互式方式输入数据库登录密码
    :param child: 交互式句柄
    :param db_user: 数据库用户
    :param db_password: 数据库密码
    :return: 登录验证状态,交互式句柄
    """
    if db_user:
        child.sendline(f'connect {db_user} as sysdba')
    else:
        child.sendline('connect sys as sysdba')
    child.expect(PexpectResult.ENTER_PASS_WORD)
    # 密码加双引号表示一个整体
    db_password = f'"{db_password}"'
    child.sendline(db_password)
    index = child.expect(PexpectResult.CONNECT_TO)
    # 0表示pexpect.TIMEOUT； 1表示pexpect.EOF
    if index in (0, 1):
        logger.error(f"Connect oracle database failed.")
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()
        return False, child

    logger.info(f"Connect oracle database success.")
    return True, child


def enter_password_asm(child, asm_password):
    """
    交互式方式输入数据库登录密码
    :param child: 交互式句柄
    :param asm_password: 数据库密码
    :return: 登录验证状态,交互式句柄
    """
    child.sendline('connect sys as sysasm')
    child.expect(PexpectResult.ENTER_PASS_WORD)
    # 密码加双引号表示一个整体
    asm_password = f'{asm_password}'
    child.sendline(asm_password)
    index = child.expect(PexpectResult.CONNECT_TO)
    # 0表示pexpect.TIMEOUT； 1表示pexpect.EOF
    if index in (0, 1):
        logger.error(f"Connect oracle database failed.")
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()
        return False, child

    logger.info(f"Connect oracle database success.")
    return True, child


def login_database(pid, db_instance_name, install_user_name, timeout):
    """
    :param pid: 任务pid
    :param db_instance_name:  数据库实例名称
    :param install_user_name: 数据库安装用户名
    :param timeout: 超时时间
    :return: 登录状态；交互式的句柄
    """
    file_name = ""
    child = None
    if platform.system().lower() == "windows":
        content = f"set NLS_LANG=American_America.US7ASCII\n set ORACLE_SID={db_instance_name} \n" \
                  f"sqlplus -M \"HTML ON\" /nolog"
        file_name = f"cmd_{pid}.bat"
        write_tmp_file(content, file_name)
        child = popen_spawn.PopenSpawn(f"{PluginPathConstant.WINDOWS_TMP_PATH}/{file_name}", timeout=timeout)
        index = child.expect(PexpectResult.OS_LOGIN_RESULT)
    else:
        if not check_os_user(install_user_name):
            logger.error(f"Not exist {install_user_name} os username.")
            return False, child
        oracle_user_env = build_user_env(user_name=install_user_name, inst_name=db_instance_name)
        child = pexpect.spawn(
            f"su - {install_user_name} -c "
            f"'{oracle_user_env} sqlplus -M \"HTML ON\" /nolog'",
            encoding="utf-8", timeout=timeout)
        index = child.expect(PexpectResult.OS_LOGIN_RESULT)
    # 删掉生成的bat文件
    if platform.system().lower() == "windows":
        check_del_dir_or_file(f"{PluginPathConstant.WINDOWS_TMP_PATH}/{file_name}")
        logger.info("Delete login database bat success")

    # 0表示pexpect.TIMEOUT； 1表示pexpect.EOF
    if index in (0, 1):
        logger.error(f"Login database {install_user_name} failed.")
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()
        return False, child
    logger.info(f"Success login database os user: {install_user_name}.")
    return True, child


def login_oracle_database(pid, db_instance_name, install_user_name, db_pass_word='', timeout=600):
    """
    交互式的方式登录oracle数据库
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :param db_pass_word: 密码
    :param timeout: 超时时间
    :return: 交互式的句柄
    """
    result, child = login_database(pid, db_instance_name, install_user_name, timeout)
    if not result:
        logger.error('Login oracle database error.')
        return False, child
    if not db_pass_word:
        db_pass_word = get_env_variable(f"job_protectObject_auth_authPwd_{pid}")
    db_user = get_env_variable(f"job_protectObject_auth_authKey_{pid}")
    result, child = enter_password(child, db_user, db_pass_word)
    clear(db_pass_word)

    if not result:
        logger.error('Invalid username or password.')
        return False, child

    return True, child


def login_oracle_database_resource(pid, db_instance_name, install_user_name, timeout=600):
    """
    交互式的方式登录oracle数据库
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :param timeout: 超时时间
    :return: 交互式的句柄
    """
    result, child = login_database(pid, db_instance_name, install_user_name, timeout)
    if not result:
        logger.error('Login oracle database error.')
        return False, child
    db_pass_word = get_env_variable(f'{OracleResourceKeyName.LIST_APPLICATION_AUTH_AUTHPWD}{pid}')
    db_user = get_env_variable(f'{OracleResourceKeyName.LIST_APPLICATION_AUTH_AUTHKEY}{pid}')
    result, child = enter_password(child, db_user, db_pass_word)
    clear(db_pass_word)

    if not result:
        logger.error('Invalid username or password.')
        return False, child

    return True, child


def login_asm_instance(pid, asm_instance_name, asm_install_user_name, asm_pass_word='', timeout=300):
    """
    交互式的方式登录asm实例
    :param pid: pid
    :param asm_instance_name: asm实例名
    :param asm_install_user_name: asm安装用户名
    :param asm_pass_word: 密码
    :param timeout: 超时时间
    :return: 交互式的句柄
    """
    result, child = login_asm(pid, asm_instance_name, asm_install_user_name, timeout)
    if not result:
        logger.error('Login oracle database error.')
        return False, child
    if not asm_pass_word and platform.system().lower() == "windows":
        asm_pass_word = get_env_variable(f"application_auth_authPwd_{pid}")
    logger.info(f'login_asm_instance asm_pass_word {asm_pass_word}')
    result, child = enter_password_asm(child, asm_pass_word)
    clear(asm_pass_word)

    if not result:
        logger.error('Invalid username or password.')
        return False, child

    return True, child


def login_asm(pid, asm_instance_name, asm_install_user_name, timeout):
    """
    :param pid: 任务pid
    :param asm_instance_name:  asm实例名称
    :param asm_install_user_name: asm安装用户名
    :param timeout: 超时时间
    :return: 登录状态；交互式的句柄
    """
    file_name = ""
    child = None
    if platform.system().lower() == "windows":
        content = f"set NLS_LANG=American_America.US7ASCII\n set ORACLE_SID={asm_instance_name} " \
                  f"\nsqlplus -M \"HTML ON\" /nolog"
        file_name = f"cmd_{pid}.bat"
        write_tmp_file(content, file_name)
        child = popen_spawn.PopenSpawn(f"{PluginPathConstant.WINDOWS_TMP_PATH}/{file_name}", timeout=timeout)
        index = child.expect(PexpectResult.OS_LOGIN_RESULT)
    else:
        if not check_os_user(asm_install_user_name):
            logger.error(f"Not exist {asm_install_user_name} os username.")
            return False, child
        asm_user_env = get_asm_user_env_by_linux(asm_install_user_name, asm_instance_name)
        child = pexpect.spawn(
            f"su - {asm_install_user_name} -c "
            f"'{asm_user_env} sqlplus -M \"HTML ON\" /nolog'",
            encoding="utf-8", timeout=timeout)
        index = child.expect(PexpectResult.OS_LOGIN_RESULT)
    # 删掉生成的bat文件
    if platform.system().lower() == "windows":
        check_del_dir_or_file(f"{PluginPathConstant.WINDOWS_TMP_PATH}/{file_name}")
        logger.info("Delete login database bat success")

    # 0表示pexpect.TIMEOUT； 1表示pexpect.EOF
    if index in (0, 1):
        logger.error(f"Login asm {asm_install_user_name} failed.")
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()
        return False, child
    logger.info(f"Success login asm os user: {asm_install_user_name}.")
    return True, child


def contains_success(child, success_patterns):
    logger.info(f"execute msg:{child.before}, after:{child.after}, error:{child.buffer}")
    compiled_patterns = [re.compile(pattern) for pattern in success_patterns if isinstance(pattern, str)]
    for child_out in (child.before, child.after, child.buffer):
        child_msg = child_out
        if isinstance(child_msg, bytes):
            child_msg = child_msg.decode(encoding="utf-8")
        logger.info(f"child_msg:{child_msg}")
        for pattern in compiled_patterns:
            if re.search(pattern, child_msg):
                logger.info(f"child_msg:{child_msg},pattern:{pattern}")
                return True, child.after
    return False, child.buffer


def exec_sql_cmd(child, sql_cmd, expect_result=None):
    """
    oracle数据库返回的查询结果
    :param expect_result: 期望返回值
    :param child: 交互式模式的句柄
    :param sql_cmd: 执行的sql语句
    :return: 是否成功执行sql语句；以HTML的形式返回数据库查询结果
    """
    logger.info(f"Executing SQL command: {sql_cmd}")
    # 合并预期结果和错误模式
    success_patterns = expect_result or PexpectResult.OPEN_STATUS
    if "select" in str(sql_cmd).lower():
        full_patterns = success_patterns + ERROR_SQL_EXPECT
    else:
        full_patterns = success_patterns + ["not logged on"]
    # 发送命令并等待响应
    child.sendline(sql_cmd)
    pattern_idx = child.expect(full_patterns)
    logger.info(f"pattern_idx:{pattern_idx}")
    if pattern_idx in (0, 1):
        error_type = full_patterns[pattern_idx]
        logger.error(f"SQL execution failed,CMD: {sql_cmd},Error: {error_type},Output: {child.before},{child.after}")
        return False, child.before
    # 解析匹配结果
    error_patterns_start = len(success_patterns)
    if pattern_idx >= error_patterns_start:
        # 匹配到错误模式（ERROR_SQL_EXPECT 范围内）
        ret, msg = contains_success(child, success_patterns)
        if ret:
            logger.info(f"return ret:{ret},msg:{msg}")
            return True, child.buffer
        error_type = full_patterns[pattern_idx]
        logger.error(f"SQL execution failed,CMD: {sql_cmd},Error: {error_type},Output: {child.before},{child.after}")
        return False, child.before
    # 成功匹配预期结果
    logger.info(f"SQL executed successfully: {sql_cmd}")
    return True, child.after


def check_database_status(pid, db_instance_name, install_user_name, primary_flag):
    """
    查询数据库状态
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :return: 开启/未开启
    """
    logger.info(f"Execute check database status, pid: {pid}, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}.")

    cmd = f"select instance_name,status from v$instance;"
    result, child = login_oracle_database(pid, db_instance_name, install_user_name)
    if not result:
        return False
    try:
        return_code, std_out = exec_sql_cmd(child, cmd)
        if return_code:
            parse_result = parse_sql_result(std_out, "STATUS")
            logger.info(f'The database status is: {parse_result}.')
            if primary_flag:
                return parse_result == InstanceStatus.OPEN.value
            else:
                return parse_result == InstanceStatus.OPEN.value or parse_result == InstanceStatus.MOUNTED.value
        logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}.")
        return False
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()


def shutdown_database_abort(pid, db_instance_name, install_user_name, db_pass_word=''):
    """
    以abort方式关闭数据库
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :param db_pass_word: 数据库登录密码
    """
    logger.info(f"Start shutdown abort, pid: {pid}, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}.")

    cmd = f"shutdown abort;"
    params = build_params(db_instance_name, install_user_name, pid)
    result, child = login_oracle_database_with_params(pid, params, timeout=30)
    if not result:
        return False
    try:
        return_code, std_out = exec_sql_cmd(child, cmd, PexpectResult.OS_LOGIN_RESULT)
        if return_code:
            return True
        logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}.")
        return False
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()


def build_params(db_instance_name, install_user_name, pid):
    return {
        'instance_name': db_instance_name,
        'db_install_user': install_user_name,
        "auth_pwd": f"job_copies_0_protectObject_auth_authPwd_{pid}",
        "auth_key": f"job_copies_0_protectObject_auth_authKey_{pid}"
    }


def shutdown_database_immediate(pid, db_instance_name, install_user_name):
    """
    以shutdown immediate方式关闭数据库
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    """
    logger.info(f"Start shutdown immediate, pid: {pid}, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}.")
    cmd = f"shutdown immediate;"
    params = build_params(db_instance_name, install_user_name, pid)
    result, child = login_oracle_database_with_params(pid, params, timeout=30)
    if not result:
        logger.error("login oracle database failure.")
        return False
    try:
        return_code, std_out = exec_sql_cmd(child, cmd, PexpectResult.SHUTDOWN)
        if return_code:
            logger.info("shutdown immediate success.")
            return True
        logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}.")
        return False
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()


def query_backup_copy_size(pid, db_instance_name, install_user_name, backup_type):
    """
    查询备份数据量大小
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :param backup_type: 备份类型
    :return: 备份数据量大小
    """
    if backup_type == BackupTypeEnum.LOG_BACKUP.value:
        command_id = "ProtectAgent_Archive_Backup"
    else:
        command_id = "ProtectAgent_Backup"

    cmd = f"set linesize 300;\n" \
          f"COL OUTPUT_BYTES_PER_SEC_DISPLAY FORMAT a50;\n" \
          f"COL SIZE_KB FORMAT 999999999999;\n" \
          f"select p.* from(select OUTPUT_BYTES_PER_SEC_DISPLAY, ROUND(OUTPUT_BYTES/1024,0) SIZE_KB from " \
          f"v$rman_backup_job_details where COMMAND_ID='{command_id}' order by SESSION_KEY desc)p where rownum = 1;"

    result, child = login_oracle_database(pid, db_instance_name, install_user_name)
    if not result:
        return 0
    try:
        return_code, std_out = exec_sql_cmd(child, cmd)
    except Exception as exception:
        logger.info(f"Failed to execute cmd: {cmd}, pid: {pid}, exception is: {exception}.")
        return 0
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()

    if return_code:
        parse_result = parse_sql_result(std_out, "SIZE_KB")
        logger.info(f'The backup copy size is: {parse_result}.')
        return int(parse_result)

    return 0


def query_archive_mode(pid, db_instance_name, install_user_name):
    """
    查询数据库归档模式
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :return: 归档/非归档
    """
    logger.info(f"Query database archive mode, pid: {pid}, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}.")

    cmd = f"select a.LOG_MODE, a.DBID, a.DB_UNIQUE_NAME, a.OPEN_MODE, b.INCARNATION#, b.RESETLOGS_ID from " \
          f"v$database a, v$database_incarnation b where b.STATUS='CURRENT';"
    result, child = login_oracle_database(pid, db_instance_name, install_user_name)
    if not result:
        return False
    try:
        return_code, std_out = exec_sql_cmd(child, cmd)
        if return_code:
            parse_result = parse_sql_result(std_out, "LOG_MODE")
            logger.info(f'The database mode is: {parse_result}.')
            return parse_result == ArchiveLogMode.ARCHIVELOG.value

        logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}.")
        return False
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()


def check_database_is_primary(pid, db_instance_name, install_user_name):
    """
    查询当前数据库是否是主库
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :return: 是/否
    """
    logger.info(f"Execute check database is primary, pid: {pid}, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}.")

    cmd = f"select database_role from v$database;"
    result, child = login_oracle_database(pid, db_instance_name, install_user_name)
    if not result:
        return False
    try:
        return_code, std_out = exec_sql_cmd(child, cmd)
        if return_code:
            parse_result = parse_sql_result(std_out, "DATABASE_ROLE")
            logger.info(f'The database role is: {parse_result}.')
            return parse_result == DataBaseRole.PRIMARY

        logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}.")
        return False
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()


def get_data_file_set(pid, db_instance_name, install_user_name):
    logger.info(f"Start get data file number set, pid: {pid}, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}.")

    result, child = login_oracle_database(pid, db_instance_name, install_user_name)
    if not result:
        raise Exception(f'Login database failed.')

    cmd = f"select file# from v$datafile;"
    try:
        exec_status, std_out = exec_sql_cmd(child, cmd)
    except Exception as exception:
        logger.error(f'Get data file failed. {str(exception)}')
        raise Exception(f'Exec {cmd} failed.') from exception
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()

    if exec_status:
        data_file_list = parse_html_result(std_out)
        data_file_num_set = set()
        for data_file in data_file_list:
            if data_file:
                data_file_num_set.add(data_file.get('FILE#'))
        return data_file_num_set

    logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}.")
    return set()


def get_oracle_file_paths(pid, db_instance_name, install_user_name, file_type):
    logger.info(f"Start get oracle file paths, pid: {pid}, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}, file type: {file_type}.")
    result, child = login_oracle_database(pid, db_instance_name, install_user_name)
    if not result:
        raise Exception(f'Login database failed.')
    try:
        cmd = OracleFilePathSql[file_type]
        exec_status, std_out = exec_sql_cmd(child, cmd, PexpectResult.FILE_PATH)
    except Exception as exception:
        logger.error(f'exec cmd failed. {str(exception)}')
        raise Exception(f'exec sql cmd failed.') from exception
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()
    if isinstance(std_out, bytes):
        std_out = std_out.decode('utf-8')
    if exec_status:
        if PexpectResult.FILE_PATH[3] in std_out:
            logger.warn(f"Failed to find file path: {cmd}, pid: {pid}.")
            return set()
        path_list = parse_html_result(std_out)
        path_set = set()
        for path in path_list:
            if path:
                path_set.update(path.values())
        logger.info(f"path_set:{path_set}")
        return path_set
    logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}.")
    return set()


def exist_first_backup(file_path):
    """
    是否存在first_backup文件
    :param file_path: 文件路径
    :return: 存在/不存在
    """
    first_backup_success = os.path.join(file_path, "additional", "first_backup_success")
    if not os.path.exists(first_backup_success):
        logger.info(f"First backup success file does not exist.")
        return False

    logger.info(f"First backup success file exist.")
    return True


def get_database_is_same(pid, job_id, db_instance_name, install_user_name):
    """
    获取数据库id
    :param pid: pid
    :param job_id: job_id
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :return: [DBID,INCARNATION,RESETLOGS_ID]
    """
    logger.info(f"Start get database id, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}, pid: {pid}, job_id: {job_id}.")
    cmd = f"select a.LOG_MODE, a.DBID, a.DB_UNIQUE_NAME, a.OPEN_MODE, b.INCARNATION#, b.RESETLOGS_ID from " \
          f"v$database a, v$database_incarnation b where b.STATUS='CURRENT';"
    result, child = login_oracle_database(pid, db_instance_name, install_user_name)
    if not result:
        return []
    try:
        return_code, std_out = exec_sql_cmd(child, cmd)
        if return_code:
            db_info_result = []
            for key in ["DBID", "INCARNATION#", "RESETLOGS_ID"]:
                parse_result = parse_sql_result(std_out, key)
                db_info_result.append(parse_result)
            return db_info_result

        logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}, job_id: {job_id}.")
        return []
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()


def check_database_id_same(file_path, pid, job_id, db_instance_name, install_user_name):
    """
    判断当前备份的数据库与上一次是否相同
    :param file_path: 文件路径
    :param pid: pid
    :param job_id: 任务id
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :return: 相同/不相同
    """
    database_info_file = os.path.join(file_path, "additional", "dbinfo")
    if not os.path.exists(database_info_file):
        logger.info(f"Database info file is not exist.")
        return False

    logger.info(f"Database info file is exist, pid: {pid}, job_id: {job_id}.")

    if platform.system().lower() == "windows":
        encoding = locale.getdefaultlocale()[1]
        file_content = read_result_file(database_info_file, encoding=encoding)
    else:
        file_content = read_result_file(database_info_file)
    logger.info(f"Additional dbinfo file content is:{file_content}, pid: {pid}, job_id: {job_id}.")

    db_info = file_content.split(";")
    old_database_id, old_incarnation_id, old_resetlogs_id = db_info[0], db_info[-3], db_info[-2]
    cur_db_info = get_database_is_same(pid, job_id, db_instance_name, install_user_name)
    logger.info(f"Old database info: {old_database_id}, {old_incarnation_id}, {old_resetlogs_id}, "
                f"current database info is: {cur_db_info}, pid: {pid}, job_id: {job_id}.")

    return old_database_id == cur_db_info[0] and old_incarnation_id == cur_db_info[1] \
        and old_resetlogs_id == cur_db_info[2]


def mount_bind_path(src_area, des_area):
    """
    进行源路径与目标路径的绑定
    :param src_area: 下发的源路径
    :param des_area: 本地待绑定的文件夹路径
    :return: 解绑定成功/失败
    """
    # windows场景下 umount和mount需要用"\\"
    if platform.system().lower() == "windows":
        src_area = src_area + "\\"
        des_area = des_area.replace("/", "\\")
    umount_bind_path(des_area)
    if platform.system().lower() == "windows":
        if os.path.isdir(des_area) or os.path.exists(des_area):
            logger.error(f"Des_area:{des_area} is exist, mount bind failed.")
            return False

        mount_bind_cmd_list = f"cmd /c mklink /d {des_area} {src_area}".split()
        return_code = execute_mount_bind_windows(mount_bind_cmd_list)

        if not return_code:
            logger.error(f"Failed to execute mount bind path err!")
            return False
    else:
        if not os.path.isdir(des_area) and not os.path.exists(des_area):
            try:
                os.mkdir(des_area, 0x700)
            except Exception as exception:
                logger.error(f"Make directory failed, err is {exception}.")
                return False
        mount_bind_cmd = f"mount --bind {src_area} {des_area}"
        return_code, std_out, std_err = execute_cmd(mount_bind_cmd)
        if return_code != CMDResult.SUCCESS.value:
            logger.error(f"Failed to execute mount bind path err: {std_err}!")
            return False

    logger.info(f"Succeed to execute mount bind path.")
    return True


def umount_bind_path(file_path):
    """
    进行路径解绑
    :param file_path: 本地绑定的文件夹路径
    :return: 解绑定成功/失败
    """
    if not file_path:
        logger.warning(f"Unmount bind path is empty!")
        return False

    if platform.system().lower() == "windows":
        umount_bind_cmd_list = f"cmd /c rmdir /s/q {file_path}".split()
        ret, result = execute_windows_cmd("", umount_bind_cmd_list, "")
        if not ret:
            logger.warning(f"Failed to exec umount bind path: {file_path}.")
            return False

        logger.info(f"Succeed to exec umount bind path {file_path}.")
        return True

    if platform.system() == 'AIX':
        umount_bind_cmd = f"umount {file_path}"
    else:
        umount_bind_cmd = f"umount -l {file_path}"
    return_code, std_out, std_err = execute_cmd(umount_bind_cmd)
    if return_code != CMDResult.SUCCESS.value:
        logger.warning(f"Failed to exec umount bind path: {file_path}, err: {std_err}.")
        return False
    logger.info(f"Succeed to exec umount bind path {file_path}.")
    return True


def exec_rc_tool_cmd(cmd, in_path, out_path):
    cmd = f"{os.path.join(ParamConstant.BIN_PATH, 'rpctool.sh')} {cmd} {in_path} {out_path}"
    ret, out, err = execute_cmd(cmd)
    if ret != CMDResult.SUCCESS:
        logger.error(f"An error occur in execute cmd. ret:{ret} err:{err}")
        return False
    return True


def write_tmp_file(context, file_name):
    """
    写临时文件供shell脚本读取
    :param context: 要写入的内容
    :param file_name: 生成的参数文件名
    :return:
    """
    tmp_path = PluginPathConstant.TMP_PATH
    if platform.system().lower() == "windows":
        tmp_path = PluginPathConstant.WINDOWS_TMP_PATH
    file_path = os.path.join(tmp_path, file_name)
    if os.path.exists(file_path):
        check_del_dir_or_file(file_path)
    output_tmp_info(file_path, context)


def check_del_dir_or_file(path):
    # 删除目录或文件
    if not os.path.exists(path):
        return
    check_path_legal(path)
    if os.path.isdir(path):
        shutil.rmtree(path)
    if os.path.isfile(path):
        os.remove(path)


def output_tmp_info(file_path, payload):
    check_path_legal(file_path)
    flags = os.O_WRONLY | os.O_CREAT
    modes = stat.S_IWUSR | stat.S_IRUSR | stat.S_IXUSR
    with os.fdopen(os.open(file_path, flags, modes), 'w') as out_file:
        out_file.write(payload)


def check_path_legal(file_path):
    real_dir_path = os.path.realpath(file_path)
    tmp_path = PluginPathConstant.TMP_PATH
    if platform.system().lower() == "windows":
        tmp_path = PluginPathConstant.WINDOWS_TMP_PATH
    if os.getenv("SystemDrive"):
        system_drive_up = os.getenv("SystemDrive").upper()
        system_drive_down = system_drive_up.lower()
        if real_dir_path.startswith(system_drive_up) or \
                real_dir_path.startswith(system_drive_down):
            if os.path.dirname(real_dir_path) != os.path.realpath(tmp_path):
                logger.error("File path not support to write")
                raise Exception("File path not support to write")
    else:
        if os.path.dirname(real_dir_path) != os.path.realpath(tmp_path):
            logger.error("File path not support to write")
            raise Exception("File path not support to write")


def convert_error_code(job_id, sub_job_id, script_error_code, default_error_code=ErrorCode.INTERNAL_ERROR):
    error_code = exit_code_error_code_map.get(script_error_code, default_error_code)
    # 如果是sql执行出错的错误码，将错误信息读取出来直接展示到界面
    if error_code in [ErrorCode.ERROR_ORACLE_EXEC_RMAN_FAILED, ErrorCode.ORACLE_EXEC_SQL_FAILED]:
        sql_result_file = os.path.join(PluginPathConstant.AGENT_ROOT_PATH.value, 'stmp',
                                       f'result_errordetail{sub_job_id}')
        if platform.system().lower() == "windows":
            sql_result_file = f"{PluginPathConstant.WINDOWS_TMP_PATH.value}/result_errordetail_{sub_job_id}"
        try:
            if platform.system().lower() == "windows":
                encoding = locale.getdefaultlocale()[1]
                sql_result = read_result_file(sql_result_file, encoding=encoding)
            else:
                sql_result = read_result_file(sql_result_file)
        except Exception as exception:
            sql_result = ''
            logger.warning(
                f'Get sql result failed, sql_result_file:{sql_result_file}, exception:{exception}, job_id:{job_id}, '
                f'sub_job_id:{sub_job_id}.')
        return error_code.value, sql_result
    error_detail = ""
    if script_error_code == ScriptExitCode.ERROR_ORACLE_RECOVERPATH_NOT_EXIT:
        error_detail = "Recover path not exist!"
    elif script_error_code == ScriptExitCode.ERROR_ORACLE_FILE_RESTORE_COPY_FAILED:
        error_detail = "Copy file failed during restore, please check permission or check log for more information."
    return error_code.value, error_detail


def convert_error_code_only(script_error_code, default_error_code=ErrorCode.INTERNAL_ERROR):
    # 无需获取sql或rman执行错误信息
    error_code = exit_code_error_code_map.get(script_error_code, default_error_code)
    return error_code.value


def get_oracle_group(install_user_name):
    group_gid = pwd.getpwnam(install_user_name).pw_gid
    group_struct = grp.getgrgid(group_gid)
    return group_struct.gr_name


def get_log_path_by_meta_type(repositories):
    for repo in repositories:
        if repo.get('repositoryType') == RepositoryDataTypeEnum.META_REPOSITORY.value:
            path = pathlib.Path(repo.get('path')[0])
            return str(path.parent.absolute())
    return ''


def get_file_owner(file_path):
    try:
        file_stat = os.stat(file_path)
    except FileNotFoundError:
        return ''
    file_uid = file_stat.st_uid
    file_owner = pwd.getpwuid(file_uid).pw_name
    return file_owner


def call_rc_tool_mount_cmd(job_id, params):
    """
    挂载或者卸载目录
    :param job_id: 任务ID
    :param params: 挂载参数
    :return:
    """

    logger.info(f"Params: {params}, job_id: {job_id}.")
    json_str = {
        "user": params.get("install_user_name", "oracle"),
        "group": params.get("oracle_group", "oinstall"),
        "fileMode": params.get("file_mode", "0750")
    }
    json_extend_info = {
        "appType": params.get("sub_type")
    }
    if params.get("fibre_channel") != "":
        json_extend_info["fibreChannel"] = params.get("fibre_channel")

    param_json = {
        "repository": [params.get('repository')],
        "permission": json_str,
        "extendInfo": json_extend_info
    }
    param_path = os.path.join(ParamConstant.PARAM_FILE_PATH, f"param_{job_id}")
    result_path = os.path.join(ParamConstant.RESULT_PATH, f"result_{job_id}")
    ret = create_param_file(param_path, param_json, job_id)
    if not ret:
        return False

    ret = exec_rc_tool_cmd(params.get("cmd", "MountRepositoryByPlugin"), param_path, result_path)
    if not ret:
        logger.warning(f"Exec rc tool failed, job_id: {job_id}.")
        return False
    ret, _ = check_path_in_white_list(param_path)
    if not ret:
        logger.error(f"Invalid param_path.")
        return False
    ret, _ = check_path_in_white_list(result_path)
    if not ret:
        logger.error(f"Invalid result_path.")
        return False
    try:
        os.remove(param_path)
    except Exception:
        # 删除参数文件失败 不认为命令执行失败
        logger.warning(f"Remove param path failed, job_id: {job_id}.")
    try:
        os.remove(result_path)
    except Exception:
        # 删除参数文件失败 不认为命令执行失败
        logger.warning(f"Remove result path failed, job_id: {job_id}.")
    return True


def create_param_file(param_path, param_json, job_id):
    ret, _ = check_path_in_white_list(param_path)
    if not ret:
        logger.error(f"Invalid param_path.")
        return False
    if os.path.exists(param_path):
        try:
            os.remove(param_path)
        except Exception as exception_str:
            logger.error(f"Rm file failed, err: {exception_str}, job_id: {job_id}.")
            return False
    logger.info(f"Write {param_json} to {param_path}, job_id:{job_id}.")
    try:
        output_execution_result(param_path, param_json)
    except Exception:
        logger.error(f"Create file failed, path: {param_path}, job_id: {job_id}.")
        return False
    return True


@retry_when_exception(retry_times=3, delay=3, logger=logger)
def change_mount_nas_file_owner(**mount_param):
    # AIX需要去查询root用户所属的组
    root_group = get_oracle_group("root")
    data_repo_path = mount_param.get("data_repo_path", "")
    oracle_user = mount_param.get("oracle_user", "")
    oracle_group = mount_param.get("oracle_group", "")
    log_repo_path = mount_param.get("log_repo_path", "")
    job_id = mount_param.get("job_id", "")
    if not data_repo_path:
        raise Exception(f"Change owner failed. data_repo_path is empty")
    cmd_list = [
        # 不修改隐藏目录/文件的权限
        f"find {data_repo_path} -name '.?*' -prune -o -exec "
        f"chown -h {oracle_user}:{oracle_group} {{}} +",
        f"chown -R -h root:{root_group} {data_repo_path}/additional",
        f"chmod 705 {data_repo_path}/additional",
        f"chown -R -h {oracle_user}:{oracle_group} {data_repo_path}/additional/dbs",
        f"chown -R -h {oracle_user}:{oracle_group} {data_repo_path}/additional/netadmin"
    ]

    if log_repo_path:
        cmd_list.append(
            f"find {log_repo_path}  -type d -name '.*' -o -name '.' -prune -o -exec "
            f"chown -h {oracle_user}:{oracle_group} {{}} +")
    for cmd in cmd_list:
        return_code, output, err_str = execute_cmd(cmd)
        if return_code != str(ScriptExitCode.SUCCESS):
            logger.error(f"Change owner failed, cmd:{cmd}, return_code:{return_code}, err_str:{err_str}, "
                         f"job_id:{job_id}.")
            raise Exception(f"Change owner failed.")
        logger.info(f'Run command successfully: {cmd}, job_id:{job_id}.')


def asm_mount(child, disk_group):
    logger.info(f"asm_mount, {disk_group}")
    cmd = f'alter diskgroup {disk_group} mount;'
    try:
        exec_status, std_out = exec_sql_cmd(child, cmd, PexpectResult.DISKGROUP_ALTERED)
    except Exception as exception:
        logger.error(f'exec asm_mount cmd failed. {str(exception)}')
        raise Exception(f'exec asm_mount cmd failed.') from exception
    if not exec_status:
        logger.error(f'asm_mount failed, cmd: {cmd}, std_out {std_out}')
        raise Exception('asm_mount failed')


def asm_startup(child):
    logger.info(f"asm_startup")
    cmd = f'startup;'
    try:
        exec_status, std_out = exec_sql_cmd(child, cmd, PexpectResult.ASM_STARTED)
        logger.info(f"execute  asm_startup exec_status: {exec_status}, std_out: {std_out}.")
    except Exception as exception:
        logger.error(f'exec asm_startup cmd failed. {str(exception)}')
        raise Exception(f'exec asm_startup cmd failed.') from exception


def get_disk_group_set(pid, child):
    logger.info(f"get disk group, pid: {pid}")
    cmd = 'select NAME from v$asm_diskgroup;'
    try:
        exec_status, std_out = exec_sql_cmd(child, cmd)
    except Exception as exception:
        logger.error(f'exec cmd failed. {str(exception)}')
        raise Exception(f'exec sql cmd failed.') from exception
    disk_group_set = set()
    if exec_status:
        disk_groups = parse_html_result(std_out)
        for disk_group in disk_groups:
            if disk_group:
                disk_group_set.update(disk_group.values())
        logger.info(f"path_set:{disk_group_set}")
        return disk_group_set
    logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}.")
    return disk_group_set


def get_spfile_path(pid, db_instance_name, install_user_name):
    logger.info(f"get_spfile_path, pid: {pid}, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}")
    result, child = login_oracle_database(pid, db_instance_name, install_user_name, timeout=30)
    if not result:
        raise Exception(f'Login database failed.')
    cmd = 'show parameter pfile;'
    try:
        exec_status, std_out = exec_sql_cmd(child, cmd)
    except Exception as exception:
        logger.error(f'exec cmd failed. {str(exception)}')
        raise Exception(f'exec sql cmd failed.') from exception
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()
    spfile = ''
    if exec_status:
        results = parse_html_result(std_out)
        spfile = results[0].get('VALUE').replace('\r', '').replace('\n', '')
        logger.info(f"get_spfile_path :{spfile}")
        return spfile
    logger.error(f"Failed to get_spfile_path cmd: {cmd}, std_out: {std_out}.")
    return spfile


def get_asm_disk_string(pid, child):
    logger.info(f"get_asm_disk_string, pid: {pid}")
    cmd = 'show parameter asm;'
    try:
        exec_status, std_out = exec_sql_cmd(child, cmd)
    except Exception as exception:
        logger.error(f'exec cmd failed. {str(exception)}')
        raise Exception(f'exec sql cmd failed.') from exception
    asm_disk_string = ''
    if exec_status:
        results = parse_html_result(std_out)
        asm_disk_string = results[1].get('VALUE')
        logger.info(f"get_asm_disk_string asm_disk_string: {asm_disk_string}.")
        return asm_disk_string
    logger.error(f"Failed to get_asm_disk_string cmd: {cmd}, std_out: {std_out}.")
    return asm_disk_string


def modify_asm_disk_string(child, asm_disk_string):
    logger.info(f"modify_asm_disk_string")
    cmd = f"alter system set asm_diskstring='{asm_disk_string}';"
    try:
        exec_status, std_out = exec_sql_cmd(child, cmd, PexpectResult.SYSTEM_ALTERED)
    except Exception as exception:
        logger.error(f'exec modify_asm_disk_string cmd failed. {str(exception)}')
        raise Exception(f'exec modify_asm_disk_string cmd failed.') from exception
    if not exec_status:
        logger.error(f'modify_asm_disk_string failed, cmd: {cmd} std_out {std_out}')
        raise Exception('modify_asm_disk_string failed')


def modify_spfile_path(pid, db_instance_name, install_user_name, spfile_path):
    oracle_user_env = get_oracle_user_env_by_linux()
    cmd_list = [
        f"su - {install_user_name} -c "
        f"'{oracle_user_env}srvctl modify database -d {db_instance_name} -p {spfile_path}'"
    ]
    logger.info(f"modify_spfile_path, cmd_list {cmd_list}")
    return_code, std_out, std_err = execute_cmd_list(cmd_list)
    if return_code != CMDResult.SUCCESS:
        logger.error(f"modify_spfile_path failed cmd {cmd_list}, std_err {std_err}")
        raise Exception(f"modify_spfile_path failed, pid {pid}, std_err {std_err}")
    logger.info(f"modify_spfile_path success, pid {pid}")


def get_datafile_mini_scn(pid, job_id, db_instance_name, install_user_name):
    """
    获取数据库datafile_mini_scn
    :param pid: pid
    :param job_id: job_id
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :return: scn
    """
    logger.info(f"Start get datafile mini scn, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}, pid: {pid}, job_id: {job_id}.")
    cmd = f"select to_char(max(CHECKPOINT_CHANGE#)) as scn from v$datafile;"
    result, child = login_oracle_database(pid, db_instance_name, install_user_name)
    if not result:
        return ""
    try:
        return_code, std_out = exec_sql_cmd(child, cmd)
    except Exception as exception:
        logger.info(f"Failed to execute cmd: {cmd}, pid: {pid}, exception is: {exception}.")
        return ""
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()
    if return_code:
        parse_result = parse_sql_result(std_out, "SCN")
        logger.info(f'The datafile mini scn is: {parse_result}.')
        return parse_result

    logger.error(f"Failed to get datafile mini scn, cmd: {cmd}, pid: {pid}, job_id: {job_id}.")
    return ""


def get_database_node_list(params, job_id):
    pid = params.get("pid")
    instance_name = params.get("instance_name")
    db_install_user = params.get("db_install_user")
    logger.info(f"Get database node list, pid:{pid}, instance name:{instance_name}, user_name:{db_install_user}.")
    if check_database_is_primary(pid, instance_name, db_install_user):
        sql_cmd = f"select THREAD# from GV$INSTANCE where STATUS='OPEN';"
    else:
        logger.info(f"Get database node list standby.")
        sql_cmd = f"SELECT DISTINCT THREAD# from v$archived_log;"
    result, child = login_oracle_database(pid, instance_name, db_install_user)
    if not result:
        return set()
    try:
        return_code, std_out = exec_sql_cmd(child, sql_cmd)
    except Exception as exception:
        logger.info(f"Failed to execute cmd: {sql_cmd}, pid: {pid}, exception is: {exception}.")
        return set()
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()

    if return_code:
        data_file_list = parse_html_result(std_out)
        data_file_num_set = set()
        for data_file in data_file_list:
            if data_file:
                data_file_num_set.add(data_file.get('THREAD#'))
        return data_file_num_set

    logger.error(f"Failed to get  database node list, cmd: {sql_cmd}, pid: {pid}, job_id: {job_id}.")
    return set()


def get_last_log_scn(params, thread, log_is_backed_up, job_id):
    pid = params.get("pid")
    instance_name = params.get("instance_name")
    db_install_user = params.get("db_install_user")
    logger.info(f"Get last log scn, instance name:{instance_name}, db_install_user:{db_install_user}")
    if thread:
        sql_cmd = f"select to_char(max(next_change#)) from v$archived_log where {log_is_backed_up} " \
                  f"and THREAD#={thread};"
    else:
        sql_cmd = f"select to_char(max(next_change#)) from v$archived_log where {log_is_backed_up};"

    result, child = login_oracle_database(pid, instance_name, db_install_user)
    if not result:
        return ""
    try:
        return_code, std_out = exec_sql_cmd(child, sql_cmd)
    except Exception as exception:
        logger.info(f"Failed to execute cmd: {sql_cmd}, pid: {pid}, exception is: {exception}.")
        return ""
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()

    if return_code:
        from_scn = parse_sql_result(std_out, "TO_CHAR(MAX(NEXT_CHANGE#))")
        logger.info(f"Get last archive log scn is {from_scn}, pid:{pid}, job_id: {job_id}")
        return from_scn
    logger.error(f"Failed to get last archive log scn, cmd: {sql_cmd}, pid: {pid}, job_id: {job_id}.")
    return ""


def get_log_time(params, thread, log_is_backed_up, from_scn):
    pid = params.get("pid")
    instance_name = params.get("instance_name")
    db_install_user = params.get("db_install_user")
    logger.info(f"Get log time, instance name:{instance_name}, db_install_user:{db_install_user}")
    if thread:
        sql_cmd = f"select to_char(next_time, {TIME_FORMAT}) from v$archived_log where NEXT_CHANGE#={from_scn} and " \
                  f"{log_is_backed_up} and THREAD#={thread};"
    else:
        sql_cmd = f"select to_char(next_time, {TIME_FORMAT}) from v$archived_log where NEXT_CHANGE#={from_scn} and " \
                  f"{log_is_backed_up};"

    result, child = login_oracle_database(pid, instance_name, db_install_user)
    if not result:
        return ""
    try:
        return_code, std_out = exec_sql_cmd(child, sql_cmd)
    except Exception as exception:
        logger.info(f"Failed to execute cmd: {sql_cmd}, pid: {pid}, exception is: {exception}.")
        return ""
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()

    if return_code:
        log_timestamp = parse_sql_result(std_out, "TO_CHAR(NEXT_TIME,'")
        logger.info(f"Get last log time is {log_timestamp}, pid:{pid}.")
        return log_timestamp
    logger.error(f"Failed to get log time, cmd: {sql_cmd}, pid: {pid}.")
    return ""


def get_database_info(params, job_id):
    pid = params.get("pid")
    instance_name = params.get("instance_name")
    db_install_user = params.get("db_install_user")
    logger.info(f"Start get resetlogs id, database instance name: {instance_name}, "
                f""f"install user name: {db_install_user}, pid: {pid}, job_id: {job_id}.")
    cmd = f"select a.LOG_MODE, a.DBID, a.DB_UNIQUE_NAME, a.OPEN_MODE, b.INCARNATION#, b.RESETLOGS_ID from " \
          f"v$database a, v$database_incarnation b where b.STATUS='CURRENT';"
    database_info = []
    result, child = login_oracle_database_with_params(pid, params)
    if not result:
        return database_info
    try:
        return_code, std_out = exec_sql_cmd(child, cmd)
    except Exception as exception:
        logger.info(f"Failed to execute cmd: {cmd}, pid: {pid}, exception is: {exception}.")
        return database_info
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()
    if return_code:
        results = parse_html_result(std_out)
        if not results or not results[0]:
            return []
        keys = ["LOG_MODE", "DBID", "DB_UNIQUE_NAME", "OPEN_MODE", "INCARNATION#", "RESETLOGS_ID"]
        database_info = [results[0].get(key) for key in keys]
        logger.info(f'The database_info id is: {database_info}.')
        return database_info
    logger.error(f"Failed to get database_info, cmd: {cmd}, pid: {pid}, job_id: {job_id}.")
    return []


def chown_file_path_owner(file_path, db_instance_name):
    stat_info = os.stat(file_path)
    user = pwd.getpwuid(stat_info.st_uid)[0]
    if user == db_instance_name:
        logger.info(f"chown_file_path_owner already {db_instance_name}")
        return
    os.lchown(file_path, pwd.getpwnam(db_instance_name).pw_uid, pwd.getpwnam(db_instance_name).pw_gid)
    logger.info(f"chown_file_path_owner success")
    return


def write_stmp_file(file_name, content):
    """
    写临时文件供shell脚本读取
    :param file_name: 生成的参数文件名
    :param content: 要写入的内容
    :return:
    """
    tmp_path = PluginPathConstant.STMP_PATH
    if platform.system().lower() == "windows":
        tmp_path = PluginPathConstant.WINDOWS_TMP_PATH
    file_path = os.path.join(tmp_path, file_name)
    logger.info(f"file_path is {file_path}")

    flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
    modes = stat.S_IWUSR | stat.S_IRUSR | stat.S_IXUSR
    with os.fdopen(os.open(file_path, flags, modes), 'w') as out_file:
        out_file.write(content)


def list_table_space(pid, db_instance_name, install_user_name, pdb_name):
    """
    获取数据库id
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :param pdb_name: pdb名称
    :return: [DBID,INCARNATION,RESETLOGS_ID]
    """
    logger.info(f"Start get database id, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}, pid: {pid}.")
    cmd = (f"select TABLESPACE_NAME from Dba_Tablespaces where TABLESPACE_NAME !='SYSAUX' "
           f"and TABLESPACE_NAME != 'SYSTEM' and TABLESPACE_NAME != 'TEMP' and TABLESPACE_NAME != 'UNDOTBS1'"
           f"and TABLESPACE_NAME != 'USERS' and TABLESPACE_NAME != 'EXAMPLE' and TABLESPACE_NAME != 'UNDOTBS2';")
    if pdb_name:
        cmd = (f"alter session set container={pdb_name};\n"
               f"select TABLESPACE_NAME from Dba_Tablespaces where TABLESPACE_NAME !='SYSAUX' "
               f"and TABLESPACE_NAME != 'SYSTEM' and TABLESPACE_NAME != 'TEMP' and TABLESPACE_NAME != 'UNDOTBS1'"
               f"and TABLESPACE_NAME != 'USERS' and TABLESPACE_NAME != 'EXAMPLE' and TABLESPACE_NAME != 'UNDOTBS2';")
    result, child = login_oracle_database_resource(pid, db_instance_name, install_user_name, timeout=30)
    if not result:
        return []
    try:
        return_code, std_out = exec_sql_cmd(child, cmd, PexpectResult.FILE_PATH)
        if return_code:
            if PexpectResult.FILE_PATH[3] in str(std_out):
                logger.warn(f"Failed to find file path: {cmd}, pid: {pid}.")
                return []
            return parse_html_result(std_out)
        logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}.")
        return []
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()


def list_user(pid, db_instance_name, install_user_name, pdb_name, task_type=TaskType.BACKUP_SERVICE):
    """
    获取数据库用户名
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :param pdb_name: pdb名称
    :param task_type: 任务类型
    :return: [USERNAME1,USERNAME2,USERNAME3]
    """
    logger.info(f"Start get database id, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}, pid: {pid}.")
    cmd = ""
    if pdb_name:
        cmd = f"alter session set container={pdb_name};\n"
    # 过滤系统用户
    cmd += (f"select username from dba_users where username not in ('ANONYMOUS','CTXSYS','DBSNMP',"
            f"'EXFSYS','LBACSYS','MDSYS','MGMT_VIEW','OLAPSYS','OWBSYS','ORDPLUGINS','ORDSYS','OUTLN',"
            f"'SI_INFORMTN_SCHEMA','SYS','SYSMAN','SYSTEM','TSMSYS','WK_TEST','WKSYS','WKPROXY','WMSYS',"
            f"'XDB','APEX_PUBLIC_USER','DIP','FLOWS_30000','FLOWS_FILES','MDDATA','ORACLE_OCM',"
            f"'SPATIAL_CSW_ADMIN_USR','SPATIAL_WFS_ADMIN_USR','XS$NULL','APPQOSSYS','DBSFWUSER','GGSYS',"
            f"'DVSYS','DVF','GSMADMIN_INTERNAL','GSMCATUSER','SYSBACKUP','REMOTE_SCHEDULER_AGENT','GSMUSER',"
            f"'SYSRAC','OJVMSYS','AUDSYS','SYSKM','ORDDATA','SYS$UMF','SYSDG');")
    result = list_result(pid, db_instance_name, install_user_name, cmd, task_type)
    return result


def list_table_by_owner(pid, db_instance_name, install_user_name, table_param_list):
    """
    根据用户名查询所属表
    :param table_param_list: 查询参数
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :param user_name: 用户名
    :param pdb_name: pdb名称
    :return: [DBID,INCARNATION,RESETLOGS_ID]
    """
    logger.info(f"Start get database id, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}, pid: {pid}.")
    cmd = ""
    if table_param_list.pdb_name:
        cmd = f"alter session set container={table_param_list.pdb_name};\n"
    begin = (table_param_list.page - 1) * table_param_list.page_size + 1
    end = table_param_list.page * table_param_list.page_size
    cmd += (f"select * from (select ROWNUM r, t.OWNER, t.TABLE_NAME, t.TABLESPACE_NAME from dba_tables t "
            f"where OWNER = '{table_param_list.owner}' and OWNER != 'SYS') where r between {begin} and {end};")
    result = list_result(pid, db_instance_name, install_user_name, cmd, TaskType.BACKUP_SERVICE)
    return result


def list_table_space_by_table(pid, db_instance_name, install_user_name, table_name, pdb_name):
    """
    根据表空间查询所属表
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :param user_name: 用户名
    :param pdb_name: pdb名称
    :return: [DBID,INCARNATION,RESETLOGS_ID]
    """
    logger.info(f"Start get database id, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}, pid: {pid}.")
    cmd = ""
    if pdb_name:
        cmd = f"alter session set container={pdb_name};\n"
    cmd += (f"select TABLESPACE_NAME,TABLE_NAME  from dba_tables "
            f"where TABLE_NAME = '{table_name}';")
    result = list_result(pid, db_instance_name, install_user_name, cmd, TaskType.BACKUP_SERVICE)
    return result


def list_table_by_table_space(pid, db_instance_name, install_user_name, table_space, pdb_name):
    """
    获取数据库id
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :param table_space: 表空间
    :param pdb_name: pdb名称
    :return: [DBID,INCARNATION,RESETLOGS_ID]
    """
    logger.info(f"Start get database id, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}, pid: {pid}.")
    cmd = ""
    if pdb_name:
        cmd = f"alter session set container={pdb_name};\n"
    cmd += (f"select OWNER, TABLE_NAME, TABLESPACE_NAME from dba_tables "
            f"where TABLESPACE_NAME = '{table_space}' and OWNER != 'SYS';")
    result = list_result(pid, db_instance_name, install_user_name, cmd, TaskType.BACKUP_SERVICE)
    return result


def list_result(pid, db_instance_name, install_user_name, cmd, task_type=TaskType.BACKUP_SERVICE):
    if task_type == TaskType.RESOURCE_SERVICE:
        result, child = login_oracle_database_resource(pid, db_instance_name, install_user_name, timeout=30)
    else:
        result, child = login_oracle_database(pid, db_instance_name, install_user_name, timeout=30)
    if not result:
        return []
    try:
        return_code, std_out = exec_sql_cmd(child, cmd, PexpectResult.FILE_PATH)
        if return_code:
            if PexpectResult.FILE_PATH[3] in str(std_out):
                logger.warn(f"Failed to find file path: {cmd}, pid: {pid}.")
                return []
            return parse_html_result(std_out)
        logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}.")
        return []
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()


def check_cdb_is_open(pid, login_params):
    logger.info("Start get_cdb_info")
    cmd = "SELECT cdb FROM v$database;"
    result, child = login_oracle_database_with_params(pid, login_params, timeout=30)
    if not result:
        return False
    try:
        return_code, std_out = exec_sql_cmd(child, cmd)
        if return_code:
            results = parse_html_result(std_out)
            cdb = results[0].get('CDB')
            return cdb == 'YES'
        logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}.")
        return False
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()


def check_database_open_mode(pid, login_params):
    logger.info("Start check_database_open_mode")
    cmd = "select open_mode from v$database;"
    result, child = login_oracle_database_with_params(pid, login_params, timeout=30)
    if not result:
        return ''
    try:
        return_code, std_out = exec_sql_cmd(child, cmd)
        if return_code:
            results = parse_html_result(std_out)
            open_mode = results[0].get('OPEN_MODE')
            return open_mode
        logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}.")
        return ''
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()


def list_pdbs(pid, db_instance_name, install_user_name, task_type=TaskType.BACKUP_SERVICE):
    """
    获取开启状态的pdb
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :return: [PDB1,PDB2]
    """
    logger.info(f"start list pdbs.")
    cmd = "select name,open_mode from v$pdbs where open_mode = 'READ WRITE';"
    if task_type == TaskType.RESOURCE_SERVICE:
        result, child = login_oracle_database_resource(pid, db_instance_name, install_user_name, timeout=30)
    else:
        result, child = login_oracle_database(pid, db_instance_name, install_user_name, timeout=30)
    if not result:
        return []
    try:
        return_code, std_out = exec_sql_cmd(child, cmd, PexpectResult.FILE_PATH)
        if return_code:
            if PexpectResult.FILE_PATH[3] in str(std_out):
                logger.warn(f"Failed to find file path: {cmd}, pid: {pid}.")
                return []
            return parse_html_result(std_out)
        logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}.")
        return []
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()


def list_all_pdbs(pid, db_instance_name, install_user_name, task_type=TaskType.BACKUP_SERVICE):
    """
    获取数据库全部pdb
    :param pid: pid
    :param db_instance_name: 数据库实例名
    :param install_user_name: 数据库安装用户名
    :return: [PDB1,PDB2]
    """
    logger.info(f"start list pdbs.")
    cmd = "select name,open_mode from v$pdbs where name != 'PDB$SEED';"
    if task_type == TaskType.RESOURCE_SERVICE:
        result, child = login_oracle_database_resource(pid, db_instance_name, install_user_name, timeout=30)
    else:
        result, child = login_oracle_database(pid, db_instance_name, install_user_name, timeout=30)
    if not result:
        return []
    try:
        return_code, std_out = exec_sql_cmd(child, cmd, PexpectResult.FILE_PATH)
        if return_code:
            if PexpectResult.FILE_PATH[3] in str(std_out):
                logger.warn(f"Failed to find file path: {cmd}, pid: {pid}, std_out={std_out}.")
                return []
            return parse_html_result(std_out)
        logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}, std_out={std_out}.")
        return []
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()


def get_compatible_info(pid, params):
    result, child = login_oracle_database_with_params(pid, params, timeout=30)
    if not result:
        raise Exception(f'Login database failed.')
    cmd = 'show parameter compatible;'
    try:
        exec_status, std_out = exec_sql_cmd(child, cmd)
    except Exception as exception:
        logger.error(f'exec cmd failed. {str(exception)}')
        raise Exception(f'exec sql cmd failed.') from exception
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()
    compatible = ''
    if exec_status:
        results = parse_html_result(std_out)
        compatible = results[0].get('VALUE').replace('\r', '').replace('\n', '')
        logger.info(f"get_spfile_path :{compatible}")
        return compatible
    logger.error(f"Failed to get_spfile_path cmd: {cmd}, std_out: {std_out}.")
    return compatible


def check_table_exist(pid, params, user_name, table_name, pdb_name):
    """
    获取数据库id
    :param pid: pid
    :param params: 数据库认证参数
    :param user_name:用户
    :param table_name: 表名
    :return: [DBID,INCARNATION,RESETLOGS_ID]
    """
    db_instance_name = params.get("instance_name")
    install_user_name = params.get("db_install_user")
    logger.info(f"Start get database id, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}, pid: {pid}.")
    cmd = (f"select count(*) count from all_tables where owner = UPPER('{user_name}') "
           f"and TABLE_NAME = '{table_name}';")
    if pdb_name:
        cmd = (f"alter session set container={pdb_name};\n"
               f"select count(*) count from all_tables where owner = UPPER('{user_name}') "
               f"and TABLE_NAME = '{table_name}';")
    result, child = login_oracle_database_with_params(pid, params, timeout=30)
    if not result:
        return False
    try:
        return_code, std_out = exec_sql_cmd(child, cmd)
        if return_code:
            results = parse_html_result(std_out)
            compatible = results[0].get('COUNT')
            return int(compatible) > 0

        logger.error(f"Failed to execute cmd: {cmd}, pid: {pid}.")
        return False
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()


def get_instance_name_by_id(instances, target_object_id=''):
    instance = get_instance_by_id(instances, target_object_id)
    if not instance:
        return ''
    if instance.get('instance_name'):
        instance_name = instance.get('instance_name')
        if not re.search("^[^|;@&$><`'!+\\n]*$", instance_name):
            logger.error(f"The instance_name{instance_name} is invalid.", instance_name)
            raise ErrCodeException(CommErrCode.PARAMS_IS_INVALID, message="The instance_name is invalid.")
        return instance_name
    return instance.get('src_instance_name')


def get_instance_by_id(instances, target_object_id=''):
    node_id = get_host_sn()
    if not instances:
        return {}
    instance_list = json.loads(instances)
    for instance in instance_list:
        if target_object_id and instance['target_object_id'] != target_object_id:
            continue
        if instance['agent_id'] == node_id:
            return instance
    return {}


def modify_instance_name_in_p_file(instances, params, target_object_id=''):
    res = {}
    # 未指定实例名，不修改
    if not instances:
        return params
    instance_list = json.loads(instances)
    for key, value in params.items():
        tmp_key = key
        for instance in instance_list:
            if target_object_id and instance['target_object_id'] != target_object_id:
                continue
            src_instance_name = instance['src_instance_name']
            dst_instance_name = instance['instance_name']
            if not dst_instance_name:
                continue
            if key.startswith(f'{src_instance_name}.'):
                tmp_key = key.replace(f'{src_instance_name}.', f'{dst_instance_name}.', 1)
        res[tmp_key] = value
    return res


def get_recover_datafiles(child):
    logger.info(f"get recover datafiles")
    cmd = 'select file# from v$datafile_header;'
    try:
        exec_status, std_out = exec_sql_cmd(child, cmd)
    except Exception as exception:
        logger.error(f'exec cmd failed. {str(exception)}')
        raise Exception(f'exec sql cmd failed.') from exception
    files_set = set()
    if exec_status:
        files = parse_html_result(std_out)
        for file in files:
            if file:
                files_set.update(file.values())
        logger.info(f"path_set:{files_set}")
        return files_set
    logger.error(f"Failed to execute cmd: {cmd}.")
    return files_set


def recover_database_file(child, file_num):
    logger.info(f"recover_database_file")
    cmd = f"recover datafile {file_num};"
    try:
        exec_status, std_out = exec_sql_cmd(child, cmd, PexpectResult.RECOVERY_COMPLETE)
    except Exception as exception:
        logger.error(f'exec recover_database_file cmd failed. {str(exception)}')
        raise Exception(f'exec recover_database_file cmd failed.') from exception
    if not exec_status:
        logger.error(f'recover_database_file failed, cmd: {cmd} std_out {std_out}')
        raise Exception('recover_database_file failed')


def get_scn_and_time(pid, job_id, db_instance_name, install_user_name):
    logger.info(f"Start get scn, database instance name: {db_instance_name}, "
                f"install user name: {install_user_name}, pid: {pid}, job_id: {job_id}.")
    cmd = f"select dbms_flashback.get_system_change_number as scn, to_char(sysdate, {TIME_FORMAT}) as time from dual;"
    result, child = login_oracle_database(pid, db_instance_name, install_user_name)
    if not result:
        return []
    try:
        return_code, std_out = exec_sql_cmd(child, cmd)
    except Exception as exception:
        logger.info(f"Failed to execute cmd: {cmd}, pid: {pid}, exception is: {exception}.")
        return []
    finally:
        if platform.system().lower() == "windows":
            child.kill(signal.SIGTERM)
        else:
            child.close()
    if return_code:
        results = parse_html_result(std_out)
        if not results or not results[0]:
            return []
        keys = ["SCN", "TIME"]
        scn_time_info = [results[0].get(key) for key in keys]
        logger.info(f'The scn and time info is: {scn_time_info}.')
        return scn_time_info

    logger.error(f"Failed to get datafile mini scn, cmd: {cmd}, pid: {pid}, job_id: {job_id}.")
    return []


def write_cert_2_file(certification, crl, unique_id):
    cert_path = ''
    crl_path = ''
    if certification:
        cert_path = os.path.join(f'{PluginPathConstant.STMP_PATH.value}', f'{unique_id}_CERT.pem')
        write_content_to_file(cert_path, certification)
    if crl:
        crl_path = os.path.join(f'{PluginPathConstant.STMP_PATH.value}', f'{unique_id}_CRL.pem')
        write_content_to_file(crl_path, crl)
    return cert_path, crl_path


def get_current_agent_uuid(agents):
    if not agents:
        return ''
    agent_uuid = ''
    local_host = get_local_ips()
    for agent in agents:
        if agent.get('ip') in local_host:
            agent_uuid = agent.get('id')
            break
    logger.info(f"get_current_agent_uuid agent_uuid {agent_uuid}")
    return agent_uuid


def check_crsd_status(pid, asm_install_user_name):
    asm_user_env = get_asm_user_env_by_linux(asm_install_user_name)
    cmd_list = [f"su - {asm_install_user_name} -c '{asm_user_env}crsctl check crs'"]
    logger.info(f"check_crsd_status, cmd_list {cmd_list}")
    return_code, std_out, std_err = execute_cmd_list(cmd_list)
    if return_code != CMDResult.SUCCESS:
        logger.error(f"check_crsd_status failed cmd {cmd_list}")
        return False
    if std_out.__contains__("Cannot communicate with Cluster Ready Services"):
        logger.error(f"check_crsd_status offline, std_err {std_err}")
        return False
    logger.info(f"check_crsd_status online, pid {pid}")
    return True


def get_grid_home(pid, asm_install_user_name):
    oracle_mark = "/###ORACLE_MARK###/"
    asm_user_env = get_asm_user_env_by_linux(asm_install_user_name)
    cmd_list = [f"su - {asm_install_user_name} -c 'echo \"{oracle_mark}\";{asm_user_env}echo $ORACLE_HOME'"]
    logger.info(f"get_grid_home, pid {pid} cmd_list {cmd_list}")
    return_code, std_out, std_err = execute_cmd_list(cmd_list)
    logger.info(f"get_grid_home:{std_out}")
    if return_code != CMDResult.SUCCESS:
        logger.error(f"get_grid_home failed, std_out {std_out} std_err {std_err}")
        return ""
    std_out_list = str(std_out).split(oracle_mark)
    if len(std_out_list) < 1:
        logger.error(f"get_grid_home null, std_out {std_out}")
        return ""
    return std_out_list[1].strip()


def create_pfile_from_spfile_linux(pid, db_instance_name, install_user_name, backup_path, db_name):
    cmd = f"create pfile='{backup_path}/ebackup-{db_name}-pfile.ora' from spfile;"
    logger.info(f"create_pfile_from_spfile {cmd}")
    result, child = login_oracle_database(pid, db_instance_name, install_user_name, timeout=100)
    if not result:
        return False
    try:
        return_code, std_out = exec_sql_cmd(child, cmd, PexpectResult.PFILE_CREATED)
        if return_code:
            return True
        logger.error(f"Failed to execute cmd: {cmd}, std_out: {std_out}.")
        return False
    finally:
        if platform.system().lower() == Platform.WINDOWS:
            child.kill(signal.SIGTERM)
        else:
            child.close()


def log_func_time_cost(func):
    # 统计方法耗时
    def fun(*args, **kwargs):
        t = time.perf_counter()
        result = func(*args, **kwargs)
        logger.info(f'[TIME_CONSUMING] func {func.__name__} cost time:{time.perf_counter() - t:.3f} s')
        return result

    return fun


def login_oracle_database_with_params(pid, params, timeout=600):
    """
    交互式的方式登录oracle数据库
    :param pid: pid
    :param params: 包括"instance_name"(str),"db_user"(str)，"db_password"(str),
    "db_install_user"(str)，"auth_pwd(str)，"auth_key"(str)
    :param timeout: 超时时间
    :return: 交互式的句柄
    """
    instance_name = params.get("instance_name", "")
    db_user = params.get("db_user", "")
    db_password = params.get("db_password", "")
    db_install_user = params.get("db_install_user", "")
    result, child = login_database(pid, instance_name, db_install_user, timeout)
    if not result:
        logger.error('Login oracle database error.')
        return False, child
    if not db_password:
        db_user, db_password = get_db_user_and_db_password(params, pid)
    result, child = enter_password(child, db_user, db_password)
    clear(db_password)

    if not result:
        logger.error('Invalid username or password.')
        return False, child

    return True, child


def get_db_user_and_db_password(params, pid):
    auth_pwd = params.get("auth_pwd", "")
    auth_key = params.get("auth_key", "")
    db_password = get_env_variable(auth_pwd)
    db_user = get_env_variable(auth_key)
    auth_pwd = f"applications_0_auth_authPwd_{pid}"
    auth_key = f"applications_0_auth_authKey_{pid}"
    if not (db_user or db_password):
        db_password = get_env_variable(auth_pwd)
        db_user = get_env_variable(auth_key)
    return db_user, db_password


def checkout_user_and_execute_cmd(pid, os_username, cmd):
    user_shell_type = get_user_shell_type(os_username)
    try:
        if user_shell_type == 'csh':
            param = ExecFuncParam(os_user=os_username, cmd_list=[cmd],
                                  fmt_params_list=[[]],
                                  chk_exe_owner=False,
                                  treat_as_error_echo_list=["logon denied"],
                                  need_input_password_echo_list=["Password:"],
                                  shell_file="/bin/csh")
        else:
            param = ExecFuncParam(os_user=os_username, cmd_list=[cmd],
                                  fmt_params_list=[[]],
                                  chk_exe_owner=False,
                                  treat_as_error_echo_list=["logon denied"],
                                  need_input_password_echo_list=["Password:"])
        result, error = su_exec_cmd_list(param)
        logger.info(f"result:{result},error:{error}")
    except Exception as e:
        logger.error(f"Execute cmd exception! sub_job_id: {pid}, exception: {Anonymity.process(e)}")
        return ScriptExitCode.ERROR_PARAM_INVALID
    if result != CMDResult.SUCCESS.value:
        logger.error(f"Execute cmd failed! sub_job_id: {pid}, cmd: {Anonymity.process(cmd)},"
                     f" result: {result}, error: {error}")
        return ScriptExitCode.ERROR_PARAM_INVALID
    return ScriptExitCode.SUCCESS


def escape_special_chars(str_contain_special_char):
    special_char_pattern = r'[^0-9a-zA-Z]'
    return re.sub(special_char_pattern, lambda x: f'\\{x.group(0)}', str_contain_special_char)


def delete_matching_start_with_pattern_files(path, start_with_pattern):
    for root, _, files in os.walk(path):
        for file in files:
            if file.startswith(start_with_pattern):
                file_path = os.path.join(root, file)
                os.remove(file_path)
                logger.info(f"Deleted: {file_path}")


def get_dir_levels(path_str):
    from pathlib import Path
    norm_path = os.path.normpath(path_str)
    path_param = Path(norm_path)
    parts = []
    for part in path_param.parts:
        if part == path_param.anchor:  # 跳过根目录或盘符（如 '/' 或 'C:\\'）
            continue
        parts.append(part)
    return parts


def general_uuid():
    """
    兼容AIX 7.1 LT4
    @return:
    """
    try:
        import uuid
        return str(uuid.uuid4())
    except Exception as err:
        logger.warn(f"general_uuid,{err}")
        timestamp = time.time_ns() // 1000
        random_part = random.randint(0, 9999)
        custom_uuid = f"{timestamp}{random_part:04d}"
        hashed_uuid = hashlib.md5(custom_uuid.encode()).hexdigest()
        return hashed_uuid
